package com.sduept.component.message.core;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.sduept.cim.enums.OperateTypeEnum;
import com.sduept.cim.exception.PersistentException;
import com.sduept.cim.model.controlarea.HostControlArea;
import com.sduept.cim.model.core.BasePower;
import com.sduept.cim.model.core.BaseVoltage;
import com.sduept.cim.model.core.ConnectivityNode;
import com.sduept.cim.model.core.DCSys;
import com.sduept.cim.model.core.IdentifiedCimObject;
import com.sduept.cim.model.core.SubControlArea;
import com.sduept.cim.model.core.Substation;
import com.sduept.cim.model.core.Terminal;
import com.sduept.cim.model.core.VoltageLevel;
import com.sduept.cim.model.generation.GeneratingUnit;
import com.sduept.cim.model.generation.HydroGeneratingUnit;
import com.sduept.cim.model.generation.ThermalGeneratingUnit;
import com.sduept.cim.model.history.core.IdentifiedCimHistoryObject;
import com.sduept.cim.model.wires.ACLineSegment;
import com.sduept.cim.model.wires.Breaker;
import com.sduept.cim.model.wires.BusbarSection;
import com.sduept.cim.model.wires.Compensator;
import com.sduept.cim.model.wires.DCPole;
import com.sduept.cim.model.wires.DCSwitch;
import com.sduept.cim.model.wires.DcLinesegment;
import com.sduept.cim.model.wires.Disconnector;
import com.sduept.cim.model.wires.EnergyConsumer;
import com.sduept.cim.model.wires.GroundDisconnector;
import com.sduept.cim.model.wires.PowerTransformer;
import com.sduept.cim.model.wires.RectifierInverter;
import com.sduept.cim.model.wires.SynchronousMachine;
import com.sduept.cim.model.wires.TapChanger;
import com.sduept.cim.model.wires.TransformerWinding;
import com.sduept.cim.persist.service.impl.CimPersistManager;
import com.sduept.component.message.model.CimEquipmentUpdateMessage;
import com.sduept.component.message.model.EquipmentUpdateMessage;
import com.sduept.core.dao.SuperDAO;
import com.sduept.generator.ObjectId;
import com.sduept.nwld.dataserver.enums.EquipmentUpdateEnum;
import com.sduept.utils.BeanUtil;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;

@Slf4j
@Service
public class CimEuipmentUpdateService {

    @Autowired
    private SuperDAO dao;
    @Autowired
    private CimPersistManager cimPersistManager;

    private ExecutorService executor = Executors.newFixedThreadPool(5);


    @KafkaListener(topics = {"cim_incremental_update"})
    public void onMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            String msgContent = jo.getString("content");
            EquipmentUpdateMessage equipmentUpdateMessage = JSON.parseObject(msgContent, EquipmentUpdateMessage.class);
            log.info("接收到消息：cim_incremental_update" );
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    if (equipmentUpdateMessage == null) {
                        return;
                    }
                    try {
                        updateCimModelByMessage(equipmentUpdateMessage);
                    } catch (Exception e) {
                        e.printStackTrace();

                    }
                }
            });
        }
    }

    /**
     * 根据OperateObject类型 更新设备模型
     *
     * @param message
     */
    public void updateCimModelByMessage(EquipmentUpdateMessage message) {
        EquipmentUpdateEnum equipmentUpdateEnum = EquipmentUpdateEnum.val(message.getOperateObject());
        switch (equipmentUpdateEnum) {
            case PRIMARY_ACLINESEGMENT:
                handleEquipment(ACLineSegment.class, message);
                break;
            case PRIMARY_BASEPOWER:
                handleEquipment(BasePower.class, message);
                break;
            case PRIMARY_BASEVOLTAGE:
                handleEquipment(BaseVoltage.class, message);
                break;
            case PRIMARY_BREAKER:
                handleEquipment(Breaker.class, message);
                break;
            case PRIMARY_BUSBARSECTION:
                handleEquipment(BusbarSection.class, message);
                break;
            case PRIMARY_COMPENSATOR:
                handleEquipment(Compensator.class, message);
                break;
            case PRIMARY_DCLINESEGMENT:
                handleEquipment(DcLinesegment.class, message);
                break;
            case PRIMARY_DCPOLE:
                handleEquipment(DCPole.class, message);
                break;
            case PRIMARY_DCSWITCH:
                handleEquipment(DCSwitch.class, message);
                break;
            case PRIMARY_DCSYS:
                handleEquipment(DCSys.class, message);
                break;
            case PRIMARY_DISCONNECTOR:
                handleEquipment(Disconnector.class, message);
                break;
            case PRIMARY_ENERGYCONSUMER:
                handleEquipment(EnergyConsumer.class, message);
                break;
            case PRIMARY_GENERATINGUNIT:
                handleEquipment(GeneratingUnit.class, message);
                break;
            case PRIMARY_GROUNDDISCONNECTOR:
                handleEquipment(GroundDisconnector.class, message);
                break;
            case PRIMARY_HOSTCONTROLAREA:
                handleEquipment(HostControlArea.class, message);
                break;
            case PRIMARY_HYDROGENERATINGUNIT:
                handleEquipment(HydroGeneratingUnit.class, message);
                break;
            case PRIMARY_POWERTRANSFORMER:
                handleEquipment(PowerTransformer.class, message);
                break;
            case PRIMARY_SUBCONTROLAREA:
                handleEquipment(SubControlArea.class, message);
                break;
            case PRIMARY_SUBSTATION:
                handleEquipment(Substation.class, message);
                break;
            case PRIMARY_RECTIFIERINVERTER:
                handleEquipment(RectifierInverter.class, message);
                break;
            case PRIMARY_SYNCHRONOUSMACHINE:
                handleEquipment(SynchronousMachine.class, message);
                break;
            case PRIMARY_TAPCHANGER:
                handleEquipment(TapChanger.class, message);
                break;
            case PRIMARY_THERMALGENERATINGUNIT:
                handleEquipment(ThermalGeneratingUnit.class, message);
                break;
            case PRIMARY_TRANSFORMERWINDING:
                handleEquipment(TransformerWinding.class, message);
                break;
            case PRIMARY_VOLTAGELEVEL:
                handleEquipment(VoltageLevel.class, message);
                break;
            case PRIMARY_TERMINAL:
                handleEquipment(Terminal.class, message);
                break;
            case PRIMARY_CONNECTIVITYNODE:
                handleEquipment(ConnectivityNode.class, message);
                break;
            default:
                break;
        }

    }

    /**
     * 模型持久化
     *
     * @param cla
     * @param message
     * @param <T>
     */
    private <T extends IdentifiedCimObject> void handleEquipment(Class<T> cla, EquipmentUpdateMessage message) {

        if (message.getOperateType().startsWith("batch")) {
            handlerBatchEquipments(cla, message);
        } else {
            handlerSingleEquipment(cla, message);
        }

    }


    /**
     * 单个模型持久化
     *
     * @param cla
     * @param message
     * @param <T>
     */
    private <T extends IdentifiedCimObject> void handlerSingleEquipment(Class<T> cla, EquipmentUpdateMessage message) {

        T bean = JSON.parseObject(message.getMessageContent(), cla);
        EquipmentUpdateEnum equipmentUpdateEnum = EquipmentUpdateEnum.val(message.getOperateType());
        List<T> cimEquipmentUpdateMessageList = new ArrayList<>();
        switch (equipmentUpdateEnum) {
            case OPERATE_CREATE:
                // 新增别称
                bean.setAliasName(bean.getName());
                try {
                    cimPersistManager.create(bean);
                } catch (PersistentException e) {
                    e.printStackTrace();
                }
                cimEquipmentUpdateMessageList.add(bean);
                saveCimEquipmentUpdateMessage(0, cimEquipmentUpdateMessageList, message);
                break;
            case OPERATE_UPDATE:
                T source = dao.findById(bean.getId(), cla);
                if (null == source) {
                    bean.setAliasName(bean.getName());
                    try {
                        cimPersistManager.update(bean);
                    } catch (PersistentException e) {
                        log.error("cim模型更新失败" + JSON.toJSONString(bean));
                        e.printStackTrace();
                    }
                } else {
                    try {
                        BeanUtil.copyBean(bean, source);
                        IdentifiedCimObject tem = (IdentifiedCimObject) source;
                        try {
                            log.error("cim模型更新失败" + JSON.toJSONString(bean));
                            cimPersistManager.update(tem);
                        } catch (PersistentException e) {
                            e.printStackTrace();
                        }
                    } catch (IllegalArgumentException e) {
                        e.printStackTrace();
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                }
                cimEquipmentUpdateMessageList.add(bean);
                saveCimEquipmentUpdateMessage(1, cimEquipmentUpdateMessageList, message);
                break;
            case OPERATE_DELETE:
                IdentifiedCimObject tem = (IdentifiedCimObject) dao.findById(bean.getId(), cla);
                if (tem == null) {
                    break;
                }
                tem.setUpdateTime(message.getUpdateTime());
                cimPersistManager.delete(bean);
                try {
                    saveHistoryData(bean, "delete");
                } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
                    e.printStackTrace();
                }
                cimEquipmentUpdateMessageList.add(bean);
                saveCimEquipmentUpdateMessage(2, cimEquipmentUpdateMessageList, message);
                break;

            default:
                break;

        }
    }

    /**
     * 批量模型持久化
     *
     * @param cla
     * @param message
     * @param <T>
     */
    private <T extends IdentifiedCimObject> void handlerBatchEquipments(Class<T> cla, EquipmentUpdateMessage message) {

        List<T> targetList = JSON.parseArray(message.getMessageContent(), cla);
        EquipmentUpdateEnum equipmentUpdateEnum = EquipmentUpdateEnum.val(message.getOperateType());
        switch (equipmentUpdateEnum) {
            case OPERATE_BATCHCREATE:
                for (T target : targetList) {
                    // 新增别称
                    target.setAliasName(target.getName());
                }
                try {
                    cimPersistManager.batchCreate(targetList);
                } catch (PersistentException e) {
                    log.error("cim模型批量新增失败" + JSON.toJSONString(targetList));
                    e.printStackTrace();
                }
                saveCimEquipmentUpdateMessage(0, targetList, message);
                break;
            case OPERATE_BATCHUPDATE:
                for (T target : targetList) {
                    target.setAliasName(target.getName());
                }
                try {
                    cimPersistManager.batchUpdate(targetList);
                } catch (PersistentException e) {
                    log.error("cim模型批量更新失败" + JSON.toJSONString(targetList));
                    e.printStackTrace();
                }
                saveCimEquipmentUpdateMessage(1, targetList, message);
                break;
            case OPERATE_BATCHDELETE:
                for (T target : targetList) {
                    target.setUpdateTime(message.getUpdateTime());

                }
                try {
                    cimPersistManager.batchDelete(targetList);
                } catch (PersistentException e) {
                    log.error("cim模型批量更新失败" + JSON.toJSONString(targetList));
                    e.printStackTrace();
                }
//                for (T target : targetList) {
//                    try {
//                        saveHistoryData(target, "delete");
//
//                    } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
//                        e.printStackTrace();
//                    }
//                }
                saveCimEquipmentUpdateMessage(2, targetList, message);
                break;

            default:
                break;

        }
        log.info("模型："+message.getOperateObject()+"操作："+message.getOperateType()+" size:"+targetList.size());

    }

    /**
     * 保存设备更新消息
     *
     * @param operation  操作（0：新增，1：更新，2：删除）
     * @param bean       IdentifiedCimObject，一次设备相关信息
     * @param updateTime 更新时间
     * @return
     */
    private <T extends IdentifiedCimObject> void saveCimEquipmentUpdateMessage(int operation, List<T> modelList,
                                                                               EquipmentUpdateMessage message) {
        CimEquipmentUpdateMessage ceum = new CimEquipmentUpdateMessage();// cim模型更新消息
        ceum.setContent(message.getMessageContent());
        String operationStr = operation == 0 ? "新增" : (operation == 1 ? "更新" : "删除");
        operationStr = "EMS模型" + operationStr;
        EquipmentUpdateEnum equipmentUpdateEnum = EquipmentUpdateEnum.val(message.getOperateObject());
        int isAlarm = 0;
        String name = modelList.stream().map(e -> e.getName()).collect(Collectors.joining(","));
        switch (equipmentUpdateEnum) {
            case PRIMARY_ACLINESEGMENT:
                operationStr += "交流线路：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_BASEPOWER:
                operationStr += "基准功率：" + name;
                break;
            case PRIMARY_BASEVOLTAGE:
                operationStr += "基础电压等级：" + name;
                break;
            case PRIMARY_BREAKER:
                List<Breaker> breakerList = Arrays.asList(modelList.toArray(new Breaker[modelList.size()]));
                name = breakerList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "断路器：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_BUSBARSECTION:
                List<BusbarSection> busbarSectionList = Arrays.asList(modelList.toArray(new BusbarSection[modelList.size()]));
                name = busbarSectionList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "母线：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_COMPENSATOR:
                List<Compensator> compensatorList = Arrays.asList(modelList.toArray(new Compensator[modelList.size()]));
                name = compensatorList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "电抗器：" + name;
                break;
            case PRIMARY_DCLINESEGMENT:
                operationStr += "直流线路：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_DCPOLE:
                operationStr += "直流极控系统：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_DCSWITCH:
                List<DCSwitch> dCSwitchList = Arrays.asList(modelList.toArray(new DCSwitch[modelList.size()]));
                name = dCSwitchList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "直流开关：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_DCSYS:
                operationStr += "直流系统：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_DISCONNECTOR:
                List<Disconnector> disconnectorList = Arrays.asList(modelList.toArray(new Disconnector[modelList.size()]));
                name = disconnectorList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "刀闸：" + name;
                break;
            case PRIMARY_ENERGYCONSUMER:
                List<EnergyConsumer> energyConsumerList = Arrays.asList(modelList.toArray(new EnergyConsumer[modelList.size()]));
                name = energyConsumerList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "能量用户：" + name;
                break;
            case PRIMARY_GENERATINGUNIT:
                List<GeneratingUnit> generatingUnitList = Arrays.asList(modelList.toArray(new GeneratingUnit[modelList.size()]));
                name = generatingUnitList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "发电机组：" + name;
                break;
            case PRIMARY_GROUNDDISCONNECTOR:
                List<GroundDisconnector> groundDisconnectorList = Arrays.asList(modelList.toArray(new GroundDisconnector[modelList.size()]));
                name = groundDisconnectorList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "接地刀闸：" + name;
                break;
            case PRIMARY_HOSTCONTROLAREA:
                operationStr += "控制区域：" + name;
                break;
            case PRIMARY_HYDROGENERATINGUNIT:
                List<HydroGeneratingUnit> hydroGeneratingUnitList = Arrays.asList(modelList.toArray(new HydroGeneratingUnit[modelList.size()]));
                name = hydroGeneratingUnitList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "发电机组：" + name;
                break;
            case PRIMARY_POWERTRANSFORMER:
                List<PowerTransformer> powerTransformerList = Arrays.asList(modelList.toArray(new PowerTransformer[modelList.size()]));
                name = powerTransformerList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "变压器组：" + name;
                break;
            case PRIMARY_SUBCONTROLAREA:
                operationStr += "子控区域：" + name;
                break;
            case PRIMARY_SUBSTATION:
                operationStr += "变电站：" + name;
                isAlarm = 1;
                break;
            case PRIMARY_RECTIFIERINVERTER:
                List<RectifierInverter> rectifierInverterList = Arrays.asList(modelList.toArray(new RectifierInverter[modelList.size()]));
                name = rectifierInverterList.stream().map(e -> e.getDcPole() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "换流器：" + name;
                break;
            case PRIMARY_SYNCHRONOUSMACHINE:
                List<SynchronousMachine> synchronousMachineList = Arrays.asList(modelList.toArray(new SynchronousMachine[modelList.size()]));
                name = synchronousMachineList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "同步电机：" + name;
                break;
            case PRIMARY_TAPCHANGER:
                List<TapChanger> tapChangerList = Arrays.asList(modelList.toArray(new TapChanger[modelList.size()]));
                name = tapChangerList.stream().map(e -> e.getTransformerWinding() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "变压器调节挡位：" + name;
                break;
            case PRIMARY_THERMALGENERATINGUNIT:
                List<ThermalGeneratingUnit> thermalGeneratingUnitList = Arrays.asList(modelList.toArray(new ThermalGeneratingUnit[modelList.size()]));
                name = thermalGeneratingUnitList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "发电机组：" + name;
                break;
            case PRIMARY_TRANSFORMERWINDING:
                List<TransformerWinding> transformerWindingList = Arrays.asList(modelList.toArray(new TransformerWinding[modelList.size()]));
                name = transformerWindingList.stream().map(e -> e.getSubName() + " " + e.getName()).collect(Collectors.joining(","));
                operationStr += "变压器绕组：" + name;
                break;
            case PRIMARY_TERMINAL:
                operationStr += "导电设备的电气连接点：" + name;
                break;
            case PRIMARY_CONNECTIVITYNODE:
                operationStr += "连接点：" + name;
                break;
            default:
                break;
        }
        ceum.setDescription(operationStr);
        ceum.setStatus(0);
        ceum.setUpdateTime(message.getUpdateTime());
        ceum.setIsAlarm(isAlarm);
        ceum.setOperation(operation);
        dao.create(ceum, CimEquipmentUpdateMessage.class);
    }

    private void saveHistoryData(IdentifiedCimObject model, String methodName)
            throws ClassNotFoundException, InstantiationException, IllegalAccessException {
        String[] split = model.getClass().getName().split("model");
        String historyClassName = split[0] + "model.history" + split[1] + "History";
        Class<?> history = Class.forName(historyClassName);
        Object object = history.newInstance();

        if ("delete".equals(methodName)) {
            copy(model, object, OperateTypeEnum.DELETE.getCode(), new Date());
            existHistory(object, history, model.getId(), OperateTypeEnum.DELETE);
        }
    }

    private void copy(Object source, Object target, int dataType, Date oprateTime)
            throws IllegalArgumentException, IllegalAccessException {
        Map<String, Object> fieldMap = getSourceFiledMap(source, dataType, oprateTime);
        setAttributes(target, fieldMap);
    }

    private static final String ID = "id";

    private static final String CIM_ID = "cimId";

    private static final String OPERATION = "operateType";

    private static final String OPERATE_TIME = "operateTime";

    private Map<String, Object> getSourceFiledMap(Object source, int dataType, Date oprateTime)
            throws IllegalArgumentException, IllegalAccessException {
        List<Field> list = BeanUtil.getFieldWithParent(source.getClass());
        Map<String, Object> result = new HashMap<String, Object>();
        for (Field field : list) {
            field.setAccessible(true);
            if (field.getName().equals(BeanUtil.SERIALVERSIONUID)) {
                continue;
            }
            if (field.getName().equals(ID)) {
                Object object = field.get(source);
                result.put(CIM_ID, object);
                continue;
            }
            Object object = field.get(source);
            result.put(field.getName(), object);
        }
        result.put(ID, ObjectId.get().toString());
        result.put(OPERATION, dataType);
        result.put(OPERATE_TIME, oprateTime);
        return result;
    }

    private void setAttributes(Object target, Map<String, Object> fieldMap) {
        List<Field> list = BeanUtil.getFieldWithParent(target.getClass());
        list.forEach(t -> setFieldValue(target, t, fieldMap));
    }

    private void setFieldValue(Object target, Field field, Map<String, Object> fieldMap) {
        field.setAccessible(true);
        Object object = fieldMap.get(field.getName());
        if (null != object) {
            try {
                field.set(target, object);
            } catch (IllegalArgumentException | IllegalAccessException e) {
                e.printStackTrace();
            }
        }
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    private <T extends IdentifiedCimHistoryObject> void existHistory(Object historyModel, Class cls, String cimId,
                                                                     OperateTypeEnum operateTypeEnum) {
        dao.createOrUpdate(historyModel, cls);
    }

}
